package operator.create;

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import util.Printer;

class FromOperator {

    public static void main(String[] args) {
//        fromArray();
//        fromCallable();
//        fromFuture();
    }


    static void fromArray() {
        Observable.fromArray(new Integer[]{1, 2, 3});
    }

    //callable = 带返回值的runnable
    static void fromCallable() {
        Observable.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 1;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Printer.printThreadName();
                Printer.print("integer = " + integer);
            }
        });

        Callable<String> callable = new Callable<String>() {
            @Override
            public String call() throws Exception {
                Printer.printThreadName("call");
                return "callable";
            }
        };
        Future<String> future = Executors.newSingleThreadExecutor().submit(callable);
//        try {
//            Thread.sleep(3000);
//            Printer.print("future.get = " + future.get());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }catch (ExecutionException e){
//            e.printStackTrace();
//        }
        Observable
                .fromCallable(callable)
//                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Printer.printThreadName("accept");
                        Printer.print("s = " + s);
                    }
                });
    }

    //Future 的作用是增加了 cancel() 等方法操作 Callable，它可以通过 get() 方法来获取 Callable 返回的值
    static void fromFuture() {
        Callable<String> callable = new Callable<String>() {
            @Override
            public String call() throws Exception {
                Printer.printThreadName("future call");
                return "call from callable";
            }
        };
        Future<String> future = Executors.newSingleThreadExecutor().submit(callable);
        //subscribe时调用了future的get方法
        Observable.fromFuture(future)
//                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Printer.printThreadName("future call");
                        Printer.print("future accept s = " + s);
                    }
                });

        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Printer.printThreadName("FutureTask call");
                return "call from futureTask";
            }
        });
        Observable.fromFuture(futureTask, Schedulers.io())
                .doOnSubscribe(disposable -> futureTask.run())
                .subscribeOn(Schedulers.io())
                .subscribe(s -> {
                    Printer.printThreadName("futureTask onNext");
                    Printer.print("futureTask accept s = " + s);
                });
    }

    static void fromIterable() {
        ArrayList<Integer> intList = new ArrayList<>();
        Observable.fromIterable(intList);
    }
}
